package com.tinem.platform.module.starter;

import com.tinem.platform.module.pojo.vo.error.SystemException;
import com.tinem.platform.module.starter.config.MqttConfiguration;
import com.tinem.platform.module.starter.sdk.MessageCode;
import com.tinem.platform.module.starter.token.TokenUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author fengzhihao
 * @version v1
 * @program: platform
 * @className MqttClient
 * @description TODO
 * @site
 * @company
 * @create 2022/1/19 12:16:43
 */
@Slf4j
public class MqttClient implements ApplicationListener<ApplicationReadyEvent> {

    Map<String, MqttConfiguration.Config> config;

    @Resource
    List<MqttMessageListener> mqttMessageListenerList;
    @Resource
    TokenUtil tokenUtil;

    public MqttClient( List<MqttConfiguration.Config> config){
        this.config = config.stream().collect(Collectors.toMap(MqttConfiguration.Config::getClientId, Function.identity()));
    }

    public void connection(String clientId) throws MqttException {
        if(!clientId.toLowerCase().equals(clientId)){
            return;
        }
        MqttConfiguration.Config config = this.config.get(clientId);
        if(config == null){
            throw new SystemException(MessageCode.ERROR_COMMONS_UNKNOWN,clientId);
        }
        MemoryPersistence persistence = new MemoryPersistence();
        config.setMqttClient(new org.eclipse.paho.client.mqttv3.MqttClient(config.getBroker(), config.getMqttClientId(), persistence));
        // MQTT 连接选项
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(config.getUserName());
        connOpts.setPassword(tokenUtil.getToken(clientId).toCharArray());

        connOpts.setAutomaticReconnect(true);
        // 保留会话
        connOpts.setCleanSession(true);
        // 设置回调
        config.getMqttClient().setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                log.info("mqtt connection error",throwable);
                try {
                    config.getMqttClient().close();
                    config.setMqttClient(null);
                } catch (MqttException e) {
                    log.info("mqtt connection error",e);
                }
                for (int i = 0; i < 5; i++) {
                    try {
                        connection(clientId);
                        log.info("mqtt connection retry success:{}",i);
                        return;
                    } catch (MqttException e) {
                        long sleep = (i+1)*5*60*1000;
                        log.info("mqtt connection retry error:{}:{}",i,sleep);
                        try {
                            Thread.sleep(sleep);
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                    }
                }
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.info("deliveryComplete---------{}" , token.isComplete());
            }
        });
        // 建立连接
        log.info("Connecting to broker: {}" , config.getBroker());
        config.getMqttClient().connect(connOpts);
        mqttMessageListenerList.stream().filter(mqttMessageListener -> {
            if(mqttMessageListener.clientId() == null){
                return true;
            }else{
                return mqttMessageListener.clientId().equals(clientId);
            }
        }).forEach(mqttMessageListener -> {
            try {
                config.getMqttClient().subscribe(mqttMessageListener.topicFilter(), mqttMessageListener.pos(), new IMqttMessageListener() {
                    @Override
                    public void messageArrived(String topic, MqttMessage message) throws Exception {
                        mqttMessageListener.messageArrived(clientId,topic,message);
                    }
                });
            } catch (MqttException e) {
                log.error("clientId:{} subscribe topic:{} error",clientId,mqttMessageListener.topicFilter(),e);
            }
        });
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        config.keySet().forEach(e->{
            try {
                this.connection(e);
            } catch (MqttException ex) {
                log.info("mett client error:{}",e,ex);
            }
        });
    }
}
